package edu.vt.ece.rt.benchmark;

import java.io.File;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

import org.jgroups.Message;

import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.messsages.types.BenchCompletion;
import edu.vt.ece.rt.quorum.messsages.types.StartNow;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.quorum.util.WaitObject;
import edu.vt.ece.rt.util.Logger.FileLogger;
import edu.vt.ece.rt.util.Logger.Logger;

public abstract class Benchmark {

	public final static int calls = Integer.getInteger(Names.calls);
	private static final Long delay=  Long.getLong(Names.transactionLength, 0);
	private static final Long TIMEOUT = Long.getLong(Names.timeOut);
	
	public static long timout(){
		return TIMEOUT + (int)(Math.random()*TIMEOUT);
	}
	
	public static void processingDelay(){
		try {
			Thread.sleep(delay);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	protected int localObjectsCount;
	protected int transactions;
	protected int readPercent;
	protected int threads;
	
	public Benchmark() {
		localObjectsCount = Integer.getInteger(Names.objects);
	}
	
	public void start() throws Throwable {
		String id = String.valueOf(ClustersManager.baseClusterId);
		File logDir = FileLogger.init(getLabel(), id);
		
		if(Boolean.getBoolean(Names.debug))
			Logger.redirect(logDir, id);

		Logger.debug("Creating Local Objects");
		
		createLocalObjects();
		ClustersManager.done = true;
		
		//Synchronize if more than 1 node
		if(Integer.getInteger(Names.nodes) > 1){
			//if root, Ask Base Quorum are then done
			if(ClustersManager.baseClusterId == 0){
				Logger.debug("Checking for publishing completion");
				ClustersManager.getCluster(Names.base).send(new Message(null, new StartNow(false, false)));
			}
			
			Logger.debug("Object creation Complete, waiting for others");
			while(ClustersManager.startnow !=true);			
		}

		long start = System.currentTimeMillis();
		transactions = Integer.getInteger(Names.transactions);
		readPercent = Integer.getInteger(Names.read);
		threads = Integer.getInteger(Names.threads);
		
		BenchmarkThread[] testingThreads = new BenchmarkThread[threads];
		
		//Create a new node Killer thread in parallel to all other benchmark thread

		for (int i=0; i<testingThreads.length; i++)
			testingThreads[i] = new BenchmarkThread(i);
		
		for (BenchmarkThread thread : testingThreads)
			thread.start();
		for (BenchmarkThread thread : testingThreads)
			thread.join();
		
		int reads=0, writes=0;
		for (BenchmarkThread thread : testingThreads){
			reads+=thread.reads;
			writes+=thread.writes;
		}
		
		String header = 
			"  n="+ClustersManager.nodesCount() + 
			", t="+threads + 
			", o="+localObjectsCount +
			", x="+transactions + 
			", c="+calls + 
			", %="+readPercent +
			", T="+Integer.getInteger(Names.timeOut) +
			", C="+Integer.getInteger(Names.callCost) +
			", L="+Integer.getInteger(Names.transactionLength) +
			", K="+Integer.getInteger(Names.linkDelay) +
			".";
		
		FileLogger.out.println(getLabel() + header);
		FileLogger.out.println("Throughput: " + ((float)1000 * transactions * threads / (System.currentTimeMillis()-start)));
		FileLogger.out.println("Reads: " + reads);
		FileLogger.out.println("Writes: " + writes);
		for(String line: result())
			FileLogger.out.println(line);
		
//		for(StackTraceElement[] elements : Thread.getAllStackTraces().values())
//			System.out.println(Arrays.toString(elements));
		
		System.err.println("Complete...[" + id + "]");

		//TODO: Should it moved to QR.close()	??
		//Start waiting for others to complete
		if(ClustersManager.baseClusterId !=0){
			System.out.println("Waiting for root done");
			while(ClustersManager.waitTillDone.rootDone.get() != true){
				System.out.print("- ");
				Thread.sleep(100);
			}			
		}else{//If you are root send everyone root Done update
			//Set auto-receive on if you are root node
			ClustersManager.getCluster(Names.base).setAutoReceive(true);
			ClustersManager.getCluster(Names.base).send(new Message(null, new BenchCompletion(true, false)));
			//wait for root done
			while(ClustersManager.waitTillDone.rootDone.get() != true){
				Thread.sleep(100);
			}	
		}
		
		//Root is done, I am also done, now reply
		WaitObject w = ClustersManager.waitTillDone;
		synchronized (w) {
			Logger.debug("Waiting for root exit message");
			ClustersManager.getCluster(Names.base).send(new Message(w.rootAdr, new BenchCompletion(false, false)));		//Unicast back to root
			w.wait();
		}
		
		
		if(ClustersManager.baseClusterId == 0){
			//Set auto receive off, default behavior of base cluster
			ClustersManager.getCluster(Names.base).setAutoReceive(false);
		}
		
		if(Boolean.getBoolean(Names.sanity)){
			System.out.println("Sanity Check ...");
			checkSanity();
			try {		//Give everyone some time to complete check sanity
				Thread.sleep(1000);				//Increase the time, if sanity check requires talking to others too
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
	}	
	
	protected String[] result() {
		return new String[] {
				"Aborts: " + Context.aborts,
//				"Conflicts: " + ContentionPolicy.dist_conflicts,
//				"Local Conflicts: " + ContentionPolicy.conflicts,
		};
	}
	
	protected void configure(String args[]){}
	
	abstract protected Object randomId();
	abstract protected int getOperandsCount();
	
	abstract protected void readOperation(Object... ids);
	abstract protected void writeOperation(Object... ids);
	abstract protected void checkSanity();

	abstract protected void createLocalObjects();
	
	abstract protected String getLabel();
	
	class BenchmarkThread extends Thread{
		int reads;
		int writes;
		
		public BenchmarkThread(int i) {
			super("Bench_"+i);
		}
		
		@Override
		public void run() {
			Logger.debug("Started");
			Random random = new Random(this.hashCode());
			
			int operands = getOperandsCount();
			for (int i = 0; i < transactions; i++) {
				List<Object> ids = new LinkedList<Object>();
				while (ids.size() < operands) {
					Object id = randomId();
					if (!ids.contains(id))
						ids.add(id);
				}
				Collections.shuffle(ids);
				boolean read = random.nextInt(100) < readPercent;
				Logger.debug((read? "R" : "W") + Arrays.toString(ids.toArray()));
				Logger.debug("start");
				try {
					long start = System.currentTimeMillis();
					if (read) {
						reads++;
						readOperation(ids.toArray());
					} else {
						writes++;
						writeOperation(ids.toArray());
					}
//					Logger.info("[" + i + "] \t" + ContextDelegator.getInstance().getRetries()  + "\t " + (System.currentTimeMillis() - start) + " \t" + Arrays.toString(ids.toArray()));
					System.err.print(".");
				} catch (Exception e) {
					e.printStackTrace();
				}
				Logger.debug("end");
//				try {
//					Thread.sleep((ClustersManager.baseClusterId+1)*100);
//				} catch (InterruptedException e) {
//					e.printStackTrace();
//				}
			}
		}
	}

}
